src/backend/tests/unit/alembic/test_shared_policy_bundle_migration.py (1)

222-244: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Add negative tests for the migration's fail-closed guards.

The three tests cover the success and repair paths. The two RuntimeError guards in the migration have no coverage:

• upgrade raises "Shared policy bundle schema is partially initialized with durable data" when exactly one table exists and it holds rows.
• _seed_active_bundle raises "Active policy bundle points to a missing immutable revision" when the active pointer references an absent revision.

These guards prevent a corrupt bundle state from being treated as valid. Add a test for each so a regression cannot turn a loud failure into a silent one. The existing _create_legacy_policy_tables and migration._create_revision_table seams make both cases cheap to set up.

As per coding guidelines: "Backend test files should ... be organized with descriptive test function names, logical setup/teardown, and coverage for positive, negative, edge, and error cases."

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/backend/tests/unit/alembic/test_shared_policy_bundle_migration.py` around
lines 222 - 244, Add two negative migration tests covering the fail-closed
guards: one should create exactly one bundle table with durable rows and assert
upgrade raises the expected partial-initialization RuntimeError; the other
should create the active bundle pointer referencing a nonexistent immutable
revision and assert _seed_active_bundle raises its missing-revision
RuntimeError. Use _create_legacy_policy_tables and
migration._create_revision_table for setup, and give each test a descriptive
name.

Source: Coding guidelines

src/backend/base/langflow/alembic/versions/f7a9c2d4e6b8_add_shared_policy_bundle.py (1)

325-332: 🗄️ Data Integrity & Integration | 🔵 Trivial | 💤 Low value

Consider dropping a single leftover table during downgrade.

When only one of the two tables exists, downgrade returns without dropping it. Alembic still records the revision as reverted, so an empty policy_bundle_revision or policy_bundle_active table stays in the schema. The upgrade path repairs that state, so the impact is limited to a leftover empty table. Dropping whichever table exists makes the downgrade fully reversible.

♻️ Proposed change

 def downgrade() -> None:
     """Copy the active bundle to legacy stores before removing new tables."""
     conn = op.get_bind()
-    if not migration.table_exists(REVISION_TABLE, conn) or not migration.table_exists(ACTIVE_TABLE, conn):
-        return
-    _sync_legacy_policy(conn)
-    op.drop_table(ACTIVE_TABLE)
-    op.drop_table(REVISION_TABLE)
+    revision_exists = migration.table_exists(REVISION_TABLE, conn)
+    active_exists = migration.table_exists(ACTIVE_TABLE, conn)
+    if revision_exists and active_exists:
+        _sync_legacy_policy(conn)
+    if active_exists:
+        op.drop_table(ACTIVE_TABLE)
+    if revision_exists:
+        op.drop_table(REVISION_TABLE)

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In
`@src/backend/base/langflow/alembic/versions/f7a9c2d4e6b8_add_shared_policy_bundle.py`
around lines 325 - 332, Update downgrade() to handle partial table state:
synchronize legacy policy data only when both REVISION_TABLE and ACTIVE_TABLE
exist, then drop each of those tables independently when it exists. Remove the
early return that currently preserves a lone table, while retaining safe
existence checks before every drop.

src/lfx/src/lfx/services/manager.py (1)

536-541: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Make the load-failure path as strict as the type check.

Line 539 raises RuntimeError when the configured class has the wrong type. However, Line 512 returns silently when load_object_from_import_path returns None for POLICY_BUNDLE_SERVICE. An operator who configures a policy bundle service that fails to import then starts with the built-in allow-all bundle service and no error. Align the two paths, as done for MODEL_PROVIDER_POLICY_SERVICE.

♻️ Proposed change at Lines 512-519

         if service_class is None:
-            if service_type == ServiceType.MODEL_PROVIDER_POLICY_SERVICE:
+            if service_type in {
+                ServiceType.MODEL_PROVIDER_POLICY_SERVICE,
+                ServiceType.POLICY_BUNDLE_SERVICE,
+            }:
                 msg = (
-                    "Configured model provider policy service could not be loaded; "
-                    "refusing to start with the OSS allow-all fallback"
+                    f"Configured {service_type.value} could not be loaded; "
+                    "refusing to start with the OSS allow-all fallback"
                 )
                 raise RuntimeError(msg)
             return

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/lfx/src/lfx/services/manager.py` around lines 536 - 541, Update the
policy bundle service loading path around load_object_from_import_path so a None
result raises a RuntimeError instead of silently falling back to the built-in
allow-all service. Match the existing strict failure behavior used for
MODEL_PROVIDER_POLICY_SERVICE, while preserving the BasePolicyBundleService
subclass validation in the POLICY_BUNDLE_SERVICE branch.

src/backend/tests/unit/services/database/test_migration_downgrade.py (2)

16-22: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

The fixture uses a URL shape that __init__ never produces.

_service bypasses __init__ with object.__new__, then sets database_url to the sync form sqlite:////configured/production.db. A real DatabaseService rewrites this value through _sanitize_database_url, so the attribute holds sqlite+aiosqlite:////configured/production.db at runtime. The assertion at Line 38 therefore locks in a URL shape that production never reaches.

The stub on _current_alembic_revisions compounds this: it removes the only code in the downgrade path that calls sa.create_engine with that URL. Use the post-sanitization URL in the fixture, and add one test that exercises _current_alembic_revisions against a real in-memory SQLite database.

As per coding guidelines: "Warn when backend pytest files rely on excessive mocks that obscure what is actually being tested."

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/backend/tests/unit/services/database/test_migration_downgrade.py` around
lines 16 - 22, The _service fixture does not represent the post-sanitization
DatabaseService state and mocks away engine creation. Set database_url to the
aiosqlite-sanitized form, remove the _current_alembic_revisions stub where
appropriate, and add a test that exercises _current_alembic_revisions against a
real in-memory SQLite database so the downgrade path validates actual engine
usage.

Source: Coding guidelines

---

41-52: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Cover the multi-head and empty-revision guard cases.

The guard uses set equality, so it also refuses a database with multiple Alembic heads and a database with no alembic_version rows. The second case produces the distinct found none message. Parametrize current_revisions over {"later_revision"}, {CURRENT_REVISION, "other_head"}, and set() to lock in all three refusals.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/backend/tests/unit/services/database/test_migration_downgrade.py` around
lines 41 - 52, The test
test_explicit_downgrade_refuses_an_unexpected_database_revision should be
parametrized over current_revisions values {"later_revision"},
{CURRENT_REVISION, "other_head"}, and set(). Update the expected RuntimeError
match so the empty set asserts the distinct “found none” message, while the
single- and multi-head cases assert their corresponding revision details; keep
downgrade.assert_not_called() for every case.

src/backend/base/langflow/services/catalog_policy/service.py (2)

70-79: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win

Consider caching the derived catalog snapshot per bundle revision.

snapshot builds a new CatalogPolicySnapshot on every access. The base class calls self.snapshot inside is_component_blocked and is_template_blocked, so a loop over many component keys constructs one object per key. The underlying frozensets are shared by reference, so a memoized projection keyed on the current bundle snapshot identity removes the repeated construction without changing behavior.

♻️ Proposed memoization of the projected snapshot

     def __init__(
         self,
         database_service: DatabaseService | None,
         policy_bundle_service: BasePolicyBundleService | None = None,
     ) -> None:
         super().__init__()
         self.database_service = database_service
         self._policy_bundle_service = policy_bundle_service
         self._legacy_snapshot = CatalogPolicySnapshot()
         self._legacy_hydrated = False
+        self._projected_source: PolicyBundleSnapshot | None = None
+        self._projected_snapshot = CatalogPolicySnapshot()
         self._write_lock = asyncio.Lock()
@@
     def snapshot(self) -> CatalogPolicySnapshot:
         """Return the current immutable process-local snapshot."""
         if self._policy_bundle_service is None:
             return self._legacy_snapshot
         bundle = self._policy_bundle_service.snapshot
-        return CatalogPolicySnapshot(
-            blocked_component_keys=bundle.blocked_component_keys,
-            blocked_template_keys=bundle.blocked_template_keys,
-        )
+        # One atomic read of the published reference, then an identity check.
+        if self._projected_source is not bundle:
+            self._projected_snapshot = CatalogPolicySnapshot(
+                blocked_component_keys=bundle.blocked_component_keys,
+                blocked_template_keys=bundle.blocked_template_keys,
+            )
+            self._projected_source = bundle
+        return self._projected_snapshot

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/backend/base/langflow/services/catalog_policy/service.py` around lines 70
- 79, Memoize the derived CatalogPolicySnapshot in the snapshot property using
the current _policy_bundle_service.snapshot identity or revision as the cache
key. Reuse the cached projection while the bundle is unchanged, and rebuild and
replace it when the bundle changes; preserve the _legacy_snapshot path when no
policy bundle service exists.

---

168-196: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low value

Consider a bounded retry when the facet write loses the compare-and-swap.

This branch reads the active revision and then replaces it in the same transaction. A concurrent component edit and template edit therefore make one caller receive a 409, even though the two facets do not overlap. The endpoint does not retry, so an operator must resubmit. A bounded retry loop that re-reads the current revision and re-applies only the requested facet would keep the atomic full-bundle write while removing the avoidable conflict.

Keep the current behavior if you want every conflict to be operator-visible.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/backend/base/langflow/services/catalog_policy/service.py` around lines
168 - 196, Add a bounded retry around the read-and-replace flow in the catalog
policy update method, re-reading the latest bundle state after a
compare-and-swap conflict and reapplying only the requested component or
template facet while preserving the other facet. Keep the atomic
replace_policy_bundle_state write and return the committed snapshot/diff on
success; propagate the conflict after the retry limit is exhausted.

src/backend/tests/unit/services/test_model_provider_policy_store.py (1)

156-158: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Assert the catalog facet that this test names.

The test name states that an external provider does not block the database-owned catalog refresh. The snapshot sets blocked_component_keys={"PythonREPL"}, but no assertion checks that catalog_service observes it. Add the catalog assertion so the test covers the named behavior.

💚 Proposed assertion

     assert policy_store.apply_model_provider_policy_state(state) is True
     assert bundle_service.snapshot is snapshot
+    assert catalog_service.snapshot.blocked_component_keys == frozenset({"PythonREPL"})
     assert provider_service.approved_provider_ids == frozenset({"openai"})

As per coding guidelines: "verify the tests actually cover the new or changed behavior rather than acting as placeholders".

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/backend/tests/unit/services/test_model_provider_policy_store.py` around
lines 156 - 158, Update the test around apply_model_provider_policy_state to
assert that catalog_service observes the snapshot’s blocked component key,
including "PythonREPL". Keep the existing bundle and provider assertions
unchanged so the test verifies the catalog facet named by the test behavior.

Source: Coding guidelines

src/backend/base/langflow/api/v1/policy_bundle.py (1)

52-55: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Bound the catalog key lists like the provider list.

approved_provider_ids limits the list to 1000 entries and constrains each ID with a pattern and max_length=255. blocked_component_keys and blocked_template_keys accept an unbounded number of entries with unbounded length. _normalize_keys only rejects empty strings. An oversized payload is then persisted as JSON and hashed.

Add a list bound and a per-key length bound for parity.

♻️ Proposed bounds

+CatalogKey = Annotated[str, StringConstraints(min_length=1, max_length=255)]
+
+
 class PolicyBundleWrite(BaseModel):
     """Complete replacement guarded by the caller's observed revision."""
 
     expected_revision: int = Field(ge=1)
     approved_provider_ids: Annotated[list[ProviderId], Field(max_length=1000)]
-    blocked_component_keys: list[str]
-    blocked_template_keys: list[str]
+    blocked_component_keys: Annotated[list[CatalogKey], Field(max_length=5000)]
+    blocked_template_keys: Annotated[list[CatalogKey], Field(max_length=5000)]

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/backend/base/langflow/api/v1/policy_bundle.py` around lines 52 - 55,
Update the policy bundle model fields blocked_component_keys and
blocked_template_keys to enforce the same catalog constraints as
approved_provider_ids: cap each list at 1000 entries and limit each key to 255
characters. Preserve the existing key normalization while applying these bounds
through the field/type declarations.

src/backend/base/langflow/api/v1/catalog_policy.py (1)

51-59: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Share one revision-conflict response builder. Both changed sites construct the same 409 detail dictionary with message, expected_revision, and active_revision. A third copy exists in src/backend/base/langflow/api/v1/policy_bundle.py at lines 134-142. The shared root cause is a missing common helper, so the client-facing conflict contract can drift between endpoints.

• src/backend/base/langflow/api/v1/catalog_policy.py#L51-L59: remove _revision_conflict and import the shared builder.
• src/backend/base/langflow/api/v1/model_provider_policy.py#L147-L155: replace the inline HTTPException construction with the same shared builder.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/backend/base/langflow/api/v1/catalog_policy.py` around lines 51 - 59,
Introduce or reuse one shared revision-conflict response builder, preserving the
existing 409 detail contract with message, expected_revision, and
active_revision. In src/backend/base/langflow/api/v1/catalog_policy.py#L51-L59,
remove _revision_conflict and import the shared builder; in
src/backend/base/langflow/api/v1/model_provider_policy.py#L147-L155, replace the
inline HTTPException construction with it. Also ensure the existing builder in
src/backend/base/langflow/api/v1/policy_bundle.py#L134-L142 is the common
implementation.

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes